package org.budo.dubbo.protocol.async.repository.activemq;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;

import lombok.ToString;
import lombok.extern.slf4j.Slf4j;

/**
 * @author limingwei
 */
@ToString
@Slf4j
public class BudoActiveMQSession {
    private Destination _destination;

    private Session _session;

    private Connection _connection;

    private MessageConsumer _messageConsumer;

    private MessageProducer _messageProducer;

    public BudoActiveMQSession(Destination destination, Session session, Connection connection) {
        this._destination = destination;
        this._session = session;

        this._connection = connection;
    }

    public MessageConsumer consumer() {
        if (null != this._messageConsumer) {
            return this._messageConsumer;
        }

        synchronized (this) {
            if (null != this._messageConsumer) {
                return this._messageConsumer;
            }

            return this._messageConsumer = this.initConsumer();
        }
    }

    public MessageProducer producer() {
        if (null != this._messageProducer) {
            return this._messageProducer;
        }

        synchronized (this) {
            if (null != this._messageProducer) {
                return this._messageProducer;
            }

            return this._messageProducer = this.initProducer();
        }
    }

    public Session getSession() {
        return this._session;
    }

    public void setMessageConsumer(MessageConsumer messageConsumer) {
        this._messageConsumer = messageConsumer;
    }

    private MessageProducer initProducer() {
        try {
            MessageProducer messageProducer = this._session.createProducer(this._destination);
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
            log.info("#77 initMessageProducer messageProducer=" + messageProducer + ", this=" + this);

            return messageProducer;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    private MessageConsumer initConsumer() {
        try {
            this._connection.start(); // connection start

            MessageConsumer messageConsumer = this._session.createConsumer(this._destination);
            log.info("#90 initMessageConsumer messageConsumer=" + messageConsumer + ", this=" + this);

            return messageConsumer;
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }
    }

    public void close() {
        try {
            if (null != this._messageProducer) {
                this._messageProducer.close();
            }
        } catch (Throwable e) {
            log.error("#92 messageProducer close error, e=" + e, e);
        }

        try {
            if (null != this._messageConsumer) {
                this._messageConsumer.close();
            }
        } catch (Throwable e) {
            log.error("#100 messageConsumer close error, e=" + e, e);
        }

        try {
            if (null != this._session) {
                this._session.close();
            }
        } catch (Throwable e) {
            log.error("#108 session close error, e=" + e, e);
        }

        try {
            if (null != this._connection) {
                this._connection.close();
            }
        } catch (Throwable e) {
            log.error("#126 connection close error, e=" + e, e);
        }
    }
}